package com.xyazm.core.service.impl;

import com.xyazm.core.common.CustomExecutorService;
import com.xyazm.core.common.Segment;
import com.xyazm.core.common.SegmentBuffer;
import com.xyazm.core.common.SegmentIDCache;
import com.xyazm.core.config.SegmentConfigProperties;
import com.xyazm.core.dao.SegmentDAO;
import com.xyazm.core.domain.SegmentDO;
import com.xyazm.core.service.ISegmentIDGenService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.*;

@Slf4j
@Service
public class SegmentIDGenServiceImpl implements ISegmentIDGenService {

    @Autowired
    private SegmentConfigProperties segmentConfigProperties;

    @Autowired
    private SegmentDAO segmentDAO;
    @Autowired
    private SegmentIDCache cache;

    private final ExecutorService threadPoolExecutor = new ThreadPoolExecutor(1, Integer.MAX_VALUE,
            60L, TimeUnit.SECONDS, new SynchronousQueue<>(), new UpdateThreadFactory());

    public static class UpdateThreadFactory implements ThreadFactory {
        private static int threadInitNumber = 0;

        private static synchronized int nextThreadNum() {
            return threadInitNumber++;
        }

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "Thread-Segment-Update-" + nextThreadNum());
        }
    }

    /**
     * 3. 获取已缓存的segmentBuffer
     * 4. 如果segmentBuffer没有初始化，则初始化：
     * 4.1 double lock
     * 4.2 调用更新方法，并设置初始化
     * 4.3 异常报错
     * 5. 从buffer中获取id
     * <p>
     * 生成新的ID
     * @param bizTag 业务标识
     */
    @Override
    public Long genNewNo(String bizTag) {
        // 校验cache是否初始化
        checkCache(bizTag);
        // 获取已缓存的segmentBuffer
        SegmentBuffer buffer = cache.getValue(bizTag);

        // 多个线程并发的积压在这里，此时双缓冲的分段数据，max、step还没初始化
        // 积压在synchronized加锁这里，只有一个线程可以进去，就会完成db里的分段数据，加载到双缓冲里来
        // 后续的线程加锁进来，双缓冲已经完成了分段数据初始化
        if (!buffer.isInitOk()) {
            synchronized (buffer) {
                if (!buffer.isInitOk()) {
                    // 获取当前的使用的segment
                    Segment segment = buffer.getCurrent();
                    try {
                        log.info("Init buffer. Update bizTag:{} from db", bizTag);
                        updateSegmentFromDb(bizTag,segment);
                        buffer.setInitOk(true);
                    } catch (Exception e) {
                        log.error("Init buffer bizTag:" + bizTag + " exception",e);
                        throw new RuntimeException("init error:" + bizTag);
                    }
                }
            }
        }
        // 从segment缓冲中获取id
        return getIdFromSegmentBuffer(buffer);
    }

    /**
     * 从segment缓冲中获取id
     * @author xuyou
     * 2023-01-03 22:14
     * @param buffer
    */
    private Long getIdFromSegmentBuffer(SegmentBuffer buffer) {
        // 必须要确保这次序列号要生成
        while (true) {
            // 本次序列号的生成分成读写锁，先加读锁，可以多线程并发读
            buffer.readLock().lock();
            try {
                // 获取当前的使用的segment
                final Segment segment = buffer.getCurrent();
                // 第一次基于current segment来进行序号生成的时候,将next segment做一个段的申请分段和加载分段到缓存里来
                if (!buffer.isNextReady() && (segment.getIdle() < segmentConfigProperties.getNextInitFactor() * segment.getStep())
                        && buffer.getThreadRunning().compareAndSet(false,true)) {
                    // 基于cas操作,确保此时只有一个线程会进入到这里，来触发next segment初始化操作
                    asyncUpdate(buffer); // 异步更新操作
                }
                long value = segment.getValue().getAndIncrement();
                if (value < segment.getMax()) {
                    return value;
                }
            } finally {
                buffer.readLock().unlock();
            }

            // 如果在这里，多个线程都可能会通过一个segment获取累加的id值，>=max
            // 必须要停顿一下，把当期的segment和另外一个备用的segment进行互换
            // 对新的segment，去库里申请一个新的分段这样子，然后继续用一个新的分段去生成序号
            // 超高并发的保护机制，基于这样的一个机制，可以保护如果next segment还在异步更新中
            // 瞬间大量线程卡在这儿，必须等待next segment更新完毕了以后再去进行发segment切换，继续发号
            waitAndSleep(buffer);
            // 多个线程都会发现thread running是false的，只有一个线程可以加到写锁
            buffer.writeLock().lock();
            try {
                final Segment segment = buffer.getCurrent();
                long value = segment.getValue().getAndIncrement();
                if (value < segment.getMax()) {
                    return value;
                }
                // nextReady，就是备用的segment，此时必须要去申请一个新的分段以及加载分段数据到segment里来
                if (buffer.isNextReady()) {
                    // 切换当前的使用的segment的index
                    buffer.switchPos();
                    // 设置下一个segment为不可切换状态
                    buffer.setNextReady(false);
                } else {
                    log.error("next segments not ready!");
                    throw new RuntimeException("next segments not ready");
                }
            } finally {
                buffer.writeLock().unlock();
            }
        }
    }

    /**
     * 异步更新操作
     * @author xuyou
     * 2023-01-03 22:36
     * @param buffer
    */
    private void asyncUpdate(SegmentBuffer buffer) {
        long submitTime = System.currentTimeMillis();
        // ExecutorService customExecutorService = CustomExecutorService.getCustomExecutorService();
        threadPoolExecutor.execute(() -> {
            long executeTime = System.currentTimeMillis();
            // 获取 next segment
            Segment next = buffer.getSegments()[buffer.nextPos()];
            boolean updateOk = false;
            try {
                updateSegmentFromDb(buffer.getBizTag(), next);
                updateOk = true;
            } catch (Exception e) {
                log.warn("bizTag:{} updateSegmentFromDb exception:{}", buffer.getBizTag(), e.getMessage());
            } finally {
                long finishTime = System.currentTimeMillis();
                log.info("update segment {} from db {}。st:{}, et:{}, ft:{}", buffer.getBizTag(),
                        next, submitTime, executeTime, finishTime);
                if (updateOk) {
                    buffer.writeLock().lock();
                    buffer.setNextReady(true);
                    buffer.getThreadRunning().set(false);
                    buffer.writeLock().unlock();
                } else {
                    buffer.getThreadRunning().set(false);
                }
            }
        });


    }


    /**
     * 自旋10000次之后，睡眠100毫秒
     */
    private void waitAndSleep(SegmentBuffer buffer) {
        int roll = 0;
        while (buffer.getThreadRunning().get()) {
            roll += 1;
            if (roll > 10000) {
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                    break;
                } catch (InterruptedException e) {
                    log.warn("Thread {} Interrupted", Thread.currentThread().getName());
                    break;
                }
            }
        }
        log.info("waitAndSleep end");
    }




    /**
     * 校验cache是否初始化
     * @author xuyou
     * 2022-12-31 20:42
     * @param bizTag
    */
    private void checkCache(String bizTag) {
        // 校验cache
        if (!cache.isInitOk()) {
            throw new RuntimeException("not init");
        }
        if (!cache.containsKey(bizTag)) {
            throw new RuntimeException("not contains key:" + bizTag);
        }
    }

    /**
     * 从db中更新号段
     * @author xuyou
     * 2022-12-31 20:55
     * @param bizTag 业务标识
     * @param segment 号段
    */
    public void updateSegmentFromDb(String bizTag, Segment segment) {
        // 获取号段对应的号段缓存器
        SegmentBuffer buffer = segment.getBuffer();

        SegmentDO orderAutoNoDO = null;
        /**
         *  1. 如果没有init
         *  1.1 更新号段最大值maxId，并获取最新orderAutoNoDO
         *  1.2 将Step、MinStep 设置成db中的step
        */
        if (!buffer.isInitOk()) {
            orderAutoNoDO = segmentDAO.updateMaxIdAndGet(bizTag);
            buffer.setStep(orderAutoNoDO.getStep());
            buffer.setMinStep(orderAutoNoDO.getStep());
        } else if (buffer.getUpdateTimestamp() == 0) {
            // 如果已经初始化过了，但是第一次初始化，更新时间戳是0，此时会在这里做一次申请segment
            /**
             *  2.如果更新时间为0，表示已经初始化过了,但是第一次更新号段,此时会在这里做一次申请segment
             *  2.1 更新号段最大值maxId，并获取最新orderAutoNoDO
             *  2.2 将Step、MinStep 设置成db中的step
             *  2.3 更新UpdateTimestamp
            */
            orderAutoNoDO = segmentDAO.updateMaxIdAndGet(bizTag);
            buffer.setStep(orderAutoNoDO.getStep());
            buffer.setMinStep(orderAutoNoDO.getStep());
            buffer.setUpdateTimestamp(System.currentTimeMillis());
        } else {
            /**
             *  3. 其他
             *  3.1 动态计算步长
            *   3.2 更新db步长等
            */
            int nextStep = calculateNextStep(bizTag, buffer);
            orderAutoNoDO = segmentDAO.updateMaxIdByDynamicStepAndGet(bizTag, nextStep);
            buffer.setUpdateTimestamp(System.currentTimeMillis());
            buffer.setStep(nextStep); // 动态算出来的
            buffer.setMinStep(orderAutoNoDO.getStep()); // min step=数据库里查出来的默认step
        }

        // max=10000，step=10000，段起始值是max-step=0
        Long maxId = orderAutoNoDO.getMaxId();
        long value =  maxId - buffer.getStep();
        // value就是当前的号段的起始序列号
        segment.getValue().set(value);
        // 号段最大值
        segment.setMax(maxId);
        // 号段步长
        segment.setStep(buffer.getStep());
        log.info("updateSegmentFromDb success, bizTag: {} ", bizTag);
    }

    /**
     * 计算新的步长
     * @author xuyou
     * 2023-01-08 16:55
     * @param bizTag 业务code
     * @param buffer 缓存
    */
    private int calculateNextStep(String bizTag, SegmentBuffer buffer) {
        /**
         * 动态步长算法，成倍扩容，折半缩容：
         * 1 如果更新间隔小于间隔阈值
         * 1.1 如果 *2 大于 阈值，不处理。不小于，则step翻倍。
         * 2 如果更新时间 小于 间隔时间*2，不处理。
         * 3 如果更新时间 大于 间隔时间*2：
         * 3.1 如果step折半小于miniStep，则折半；
        */
        // 这一次更新，距离上一次更新的时间差距
        long duration = System.currentTimeMillis() - buffer.getUpdateTimestamp();
        // 拿出来当前的step，作为next step
        int nextStep = buffer.getStep();
        // 这一次距离上一次的更新时间差距 < 15分钟（阈值）
        // 如果发现两次segment更新时间，太快了，都小于15分钟了，更新太频繁，步长自适应调整
        if (duration < segmentConfigProperties.getSegmentDuration())  {
            nextStep = Math.min(segmentConfigProperties.getMaxStep(), nextStep * segmentConfigProperties.getExpandFactor()); // 10000 -> 20000
        } else if (duration < segmentConfigProperties.getSegmentDuration() * segmentConfigProperties.getExpandFactor()) {
            // 会尽可能的让你的发号时间间隔控制在15~30分钟这个范围里
            // do nothing with nextStep
        } else {
            // 如果说两次更新的时间间隔是大于了30分钟，15~30这个区间什么都没干
            // 如果说两次更新时间过长了，超过30分钟，步长太大了，发号发的很慢，此时会自动缩容，/2，10000 -> 5000
            nextStep = Math.max(buffer.getMinStep(), nextStep / segmentConfigProperties.getExpandFactor());
        }
        log.info("bizTag[{}], step[{}], duration[{}min], nextStep[{}]", bizTag, buffer.getStep(),
                String.format("%.2f", ((double) duration / (1000 * 60))), nextStep);
        return nextStep;
    }


}
